package juc.scenarios.datafetch;

import lombok.Getter;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 高级并发数据抓取系统演示
 * <p>
 * 技术点：Semaphore + 线程池 + CompletableFuture + 重试机制
 * 业务场景：企业级数据抓取系统，支持动态并发控制、智能重试、熔断保护等高级特性
 * <p>
 * 功能特点：
 * 1. 动态并发控制（根据系统负载调整）
 * 2. 智能重试机制（指数退避）
 * 3. 熔断保护（失败率过高时暂停）
 * 4. 实时监控和告警
 * 5. 数据质量检查
 * 6. 批量处理和缓存
 */
public class AdvancedDataFetchDemo {

    // 数据源配置
    private static final Map<String, DataSourceConfig> DATA_SOURCES = new HashMap<>();

    // 系统配置
    private static final int INITIAL_CONCURRENT_FETCHES = 3;
    private static final int MAX_CONCURRENT_FETCHES = 8;
    private static final int MAX_RETRIES = 3;
    private static final long CIRCUIT_BREAKER_THRESHOLD = 5; // 连续失败阈值
    private static final long CIRCUIT_BREAKER_TIMEOUT = 10000; // 熔断超时时间（毫秒）

    // 核心组件
    private final Semaphore semaphore;
    private final ExecutorService executor;
    private final ScheduledExecutorService monitorExecutor;

    // 统计和监控
    private final AtomicInteger successCount = new AtomicInteger(0);
    private final AtomicInteger failureCount = new AtomicInteger(0);
    private final AtomicLong totalFetchTime = new AtomicLong(0);
    private final AtomicInteger consecutiveFailures = new AtomicInteger(0);

    // 熔断器状态
    private volatile boolean circuitBreakerOpen = false;
    private volatile long circuitBreakerOpenTime = 0;

    // 缓存和结果存储
    private final Map<String, CompletableFuture<FetchResult>> fetchFutures = new ConcurrentHashMap<>();
    private final Map<String, FetchResult> cachedResults = new ConcurrentHashMap<>();

    static {
        // 初始化数据源配置
        DATA_SOURCES.put("用户服务API", new DataSourceConfig("用户服务API", 1, 1000, 0.1));
        DATA_SOURCES.put("订单服务API", new DataSourceConfig("订单服务API", 2, 1500, 0.15));
        DATA_SOURCES.put("商品服务API", new DataSourceConfig("商品服务API", 1, 800, 0.05));
        DATA_SOURCES.put("库存服务API", new DataSourceConfig("库存服务API", 3, 600, 0.2));
        DATA_SOURCES.put("支付服务API", new DataSourceConfig("支付服务API", 2, 1200, 0.1));
        DATA_SOURCES.put("物流服务API", new DataSourceConfig("物流服务API", 1, 2000, 0.25));
        DATA_SOURCES.put("推荐服务API", new DataSourceConfig("推荐服务API", 1, 900, 0.08));
        DATA_SOURCES.put("营销服务API", new DataSourceConfig("营销服务API", 2, 1100, 0.12));
        DATA_SOURCES.put("客服服务API", new DataSourceConfig("客服服务API", 1, 1800, 0.3));
        DATA_SOURCES.put("统计服务API", new DataSourceConfig("统计服务API", 1, 2500, 0.4));
        DATA_SOURCES.put("日志服务API", new DataSourceConfig("日志服务API", 1, 3000, 0.5));
        DATA_SOURCES.put("配置服务API", new DataSourceConfig("配置服务API", 1, 500, 0.02));
    }

    /**
     * 数据源配置
     */
    private static class DataSourceConfig {
        final String name;
        final int priority; // 优先级（1-3，数字越小优先级越高）
        final int baseFetchTime; // 基础抓取时间（毫秒）
        final double failureRate; // 模拟失败率

        DataSourceConfig(String name, int priority, int baseFetchTime, double failureRate) {
            this.name = name;
            this.priority = priority;
            this.baseFetchTime = baseFetchTime;
            this.failureRate = failureRate;
        }
    }

    /**
     * 数据抓取结果（继承自基础版本）
     */
    @Getter
    public static class FetchResult {
        // Getters
        private final String dataSource;
        private final boolean success;
        private final String data;
        private final long fetchTime;
        private final int retryCount;
        private final String error;
        private final LocalDateTime timestamp;
        private final boolean fromCache;
        private final int dataQualityScore; // 数据质量评分（0-100）

        public FetchResult(String dataSource, boolean success, String data, long fetchTime,
                           int retryCount, String error, boolean fromCache, int dataQualityScore) {
            this.dataSource = dataSource;
            this.success = success;
            this.data = data;
            this.fetchTime = fetchTime;
            this.retryCount = retryCount;
            this.error = error;
            this.timestamp = LocalDateTime.now();
            this.fromCache = fromCache;
            this.dataQualityScore = dataQualityScore;
        }

        @Override
        public String toString() {
            return String.format("%s: %s (耗时: %dms, 重试: %d次, 质量: %d)%s%s",
                    dataSource,
                    success ? "成功" : "失败",
                    fetchTime,
                    retryCount,
                    dataQualityScore,
                    fromCache ? " [缓存]" : "",
                    success ? "" : ", 错误: " + error
            );
        }
    }

    /**
     * 系统统计信息
     */
    public static class SystemStats {
        private final int totalRequests;
        private final int successCount;
        private final int failureCount;
        private final double successRate;
        private final long avgResponseTime;
        private final int activeConnections;
        private final int queueSize;
        private final boolean circuitBreakerOpen;
        private final double systemLoad;

        public SystemStats(int totalRequests, int successCount, int failureCount,
                           long avgResponseTime, int activeConnections, int queueSize,
                           boolean circuitBreakerOpen, double systemLoad) {
            this.totalRequests = totalRequests;
            this.successCount = successCount;
            this.failureCount = failureCount;
            this.successRate = totalRequests > 0 ? (double) successCount / totalRequests * 100 : 0;
            this.avgResponseTime = avgResponseTime;
            this.activeConnections = activeConnections;
            this.queueSize = queueSize;
            this.circuitBreakerOpen = circuitBreakerOpen;
            this.systemLoad = systemLoad;
        }

        @Override
        public String toString() {
            return String.format(
                    "系统统计: 总请求=%d, 成功率=%.1f%%, 平均响应=%dms, " +
                            "活跃连接=%d, 队列=%d, 熔断器=%s, 系统负载=%.1f%%",
                    totalRequests, successRate, avgResponseTime,
                    activeConnections, queueSize,
                    circuitBreakerOpen ? "开启" : "关闭",
                    systemLoad * 100
            );
        }
    }

    public AdvancedDataFetchDemo() {
        this.semaphore = new Semaphore(INITIAL_CONCURRENT_FETCHES);
        this.executor = new ThreadPoolExecutor(
                INITIAL_CONCURRENT_FETCHES,
                MAX_CONCURRENT_FETCHES * 2,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                new ThreadFactory() {
                    private final AtomicInteger threadNumber = new AtomicInteger(1);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r, "AdvancedDataFetch-Worker-" + threadNumber.getAndIncrement());
                        t.setDaemon(false);
                        return t;
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        this.monitorExecutor = Executors.newScheduledThreadPool(1, r -> {
            Thread t = new Thread(r, "DataFetch-Monitor");
            t.setDaemon(true);
            return t;
        });

        // 启动监控任务
        startMonitoring();
    }

    /**
     * 数据抓取任务（支持优先级）
     */
    private class DataFetchTask implements Callable<FetchResult>, Comparable<DataFetchTask> {
        private final String dataSource;
        private final DataSourceConfig config;
        private final int retryCount;

        public DataFetchTask(String dataSource, int retryCount) {
            this.dataSource = dataSource;
            this.config = DATA_SOURCES.get(dataSource);
            this.retryCount = retryCount;
        }

        @Override
        public FetchResult call() {
            String threadName = Thread.currentThread().getName();
            System.out.printf("[%s] 开始抓取: %s (重试: %d, 优先级: %d)%n",
                    threadName, dataSource, retryCount, config.priority);

            long startTime = System.currentTimeMillis();

            try {
                // 检查熔断器
                if (isCircuitBreakerOpen()) {
                    throw new RuntimeException("熔断器开启，暂停抓取");
                }

                // 模拟数据抓取
                int fetchTime = simulateDataFetch();

                // 数据质量检查
                int dataQualityScore = checkDataQuality();

                // 生成数据
                String data = generateMockData(dataSource);

                long endTime = System.currentTimeMillis();
                long duration = endTime - startTime;

                FetchResult result = new FetchResult(
                        dataSource, true, data, duration, retryCount, null, false, dataQualityScore
                );

                // 更新统计
                successCount.incrementAndGet();
                totalFetchTime.addAndGet(duration);
                consecutiveFailures.set(0);

                // 缓存结果
                cachedResults.put(dataSource, result);

                System.out.printf("[%s] 完成抓取: %s (耗时: %dms, 质量: %d)%n",
                        threadName, dataSource, duration, dataQualityScore);

                return result;

            } catch (Exception e) {
                long endTime = System.currentTimeMillis();
                long duration = endTime - startTime;

                // 更新失败统计
                failureCount.incrementAndGet();
                consecutiveFailures.incrementAndGet();

                // 检查是否需要开启熔断器
                checkCircuitBreaker();

                FetchResult result = new FetchResult(
                        dataSource, false, null, duration, retryCount, e.getMessage(), false, 0
                );

                System.err.printf("[%s] 抓取失败: %s (错误: %s)%n",
                        threadName, dataSource, e.getMessage());

                return result;

            } finally {
                semaphore.release();
            }
        }

        @Override
        public int compareTo(DataFetchTask other) {
            // 按优先级排序（数字小的优先级高）
            return Integer.compare(this.config.priority, other.config.priority);
        }

        /**
         * 模拟数据抓取时间（带随机性）
         */
        private int simulateDataFetch() throws InterruptedException {
            // 基础时间 + 随机延迟
            int fetchTime = config.baseFetchTime + (int) (Math.random() * 500);

            // 模拟随机失败
            if (Math.random() < config.failureRate) {
                throw new RuntimeException("服务不可用");
            }

            Thread.sleep(fetchTime);
            return fetchTime;
        }

        /**
         * 数据质量检查
         */
        private int checkDataQuality() {
            // 模拟数据质量评分（70-100分）
            return 70 + (int) (Math.random() * 30);
        }

        /**
         * 生成模拟数据
         */
        private String generateMockData(String dataSource) {
            Random random = new Random();
            StringBuilder data = new StringBuilder();
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

            data.append("{");
            data.append("\"source\":\"").append(dataSource).append("\",");
            data.append("\"timestamp\":\"").append(LocalDateTime.now().format(formatter)).append("\",");

            if (dataSource.contains("用户")) {
                data.append("\"type\":\"user\",");
                data.append("\"data\":{\"userId\":").append(random.nextInt(10000));
                data.append(",\"userName\":\"user").append(random.nextInt(1000)).append("\"}");
            } else if (dataSource.contains("订单")) {
                data.append("\"type\":\"order\",");
                data.append("\"data\":{\"orderId\":\"ORD-\"").append(random.nextInt(100000));
                data.append("\",\"amount\":").append(String.format("%.2f", random.nextDouble() * 1000));
                data.append(",\"status\":\"completed\"}");
            } else {
                data.append("\"type\":\"generic\",");
                data.append("\"data\":{\"id\":").append(random.nextInt(10000));
                data.append(",\"status\":\"active\"}");
            }

            data.append("}");
            return data.toString();
        }
    }

    /**
     * 执行高级并发数据抓取
     */
    public CompletableFuture<Map<String, FetchResult>> fetchDataConcurrently() {
        System.out.println("\n🚀 高级并发数据抓取系统启动");
        System.out.println("数据源数量: " + DATA_SOURCES.size());
        System.out.println("初始并发数: " + INITIAL_CONCURRENT_FETCHES);
        System.out.println("最大重试次数: " + MAX_RETRIES);
        System.out.println("=".repeat(60));

        long startTime = System.currentTimeMillis();

        // 为每个数据源创建抓取任务
        List<CompletableFuture<FetchResult>> allFutures = new ArrayList<>();

        DATA_SOURCES.keySet().forEach(dataSource -> {
            CompletableFuture<FetchResult> future = fetchWithRetry(dataSource, 0);
            fetchFutures.put(dataSource, future);
            allFutures.add(future);
        });

        // 等待所有任务完成
        return CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0]))
                .thenApply(v -> {
                    Map<String, FetchResult> results = new HashMap<>();

                    fetchFutures.forEach((dataSource, future) -> {
                        try {
                            FetchResult result = future.get();
                            results.put(dataSource, result);
                        } catch (Exception e) {
                            System.err.println("获取结果失败: " + e.getMessage());
                        }
                    });

                    long endTime = System.currentTimeMillis();
                    long totalTime = endTime - startTime;

                    printStatistics(totalTime);

                    return results;
                });
    }

    /**
     * 带重试机制的抓取
     */
    private CompletableFuture<FetchResult> fetchWithRetry(String dataSource, int retryCount) {
        // 检查缓存
        FetchResult cachedResult = cachedResults.get(dataSource);
        if (cachedResult != null && !isCacheExpired(cachedResult)) {
            return CompletableFuture.completedFuture(
                    new FetchResult(
                            cachedResult.getDataSource(),
                            cachedResult.isSuccess(),
                            cachedResult.getData(),
                            0,
                            retryCount,
                            null,
                            true,
                            cachedResult.getDataQualityScore()
                    )
            );
        }

        return CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();
                return new DataFetchTask(dataSource, retryCount).call();
            } catch (Exception e) {
                // 重试逻辑
                if (retryCount < MAX_RETRIES && shouldRetry(e)) {
                    long delay = calculateRetryDelay(retryCount);
                    System.out.printf("等待 %dms 后重试: %s (第 %d 次)%n", delay, dataSource, retryCount + 1);

                    try {
                        Thread.sleep(delay);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }

                    return fetchWithRetry(dataSource, retryCount + 1).join();
                } else {
                    return new FetchResult(
                            dataSource, false, null, 0, retryCount, e.getMessage(), false, 0
                    );
                }
            }
        }, executor);
    }

    /**
     * 计算重试延迟（指数退避）
     */
    private long calculateRetryDelay(int retryCount) {
        // 指数退避：100ms, 200ms, 400ms...
        return 100L * (1L << retryCount);
    }

    /**
     * 判断是否应该重试
     */
    private boolean shouldRetry(Exception e) {
        String message = e.getMessage();
        return message != null && (
                message.contains("超时") ||
                        message.contains("不可用") ||
                        message.contains("熔断器")
        );
    }

    /**
     * 检查缓存是否过期（5分钟）
     */
    private boolean isCacheExpired(FetchResult result) {
        return result.getTimestamp().isBefore(LocalDateTime.now().minusMinutes(5));
    }

    /**
     * 检查熔断器状态
     */
    private boolean isCircuitBreakerOpen() {
        if (circuitBreakerOpen) {
            // 检查是否应该关闭熔断器
            if (System.currentTimeMillis() - circuitBreakerOpenTime > CIRCUIT_BREAKER_TIMEOUT) {
                circuitBreakerOpen = false;
                consecutiveFailures.set(0);
                System.out.println("熔断器已关闭，恢复抓取");
                return false;
            }
            return true;
        }
        return false;
    }

    /**
     * 检查并控制熔断器
     */
    private void checkCircuitBreaker() {
        if (consecutiveFailures.get() >= CIRCUIT_BREAKER_THRESHOLD && !circuitBreakerOpen) {
            circuitBreakerOpen = true;
            circuitBreakerOpenTime = System.currentTimeMillis();
            System.err.println("⚠️ 熔断器已开启，暂停数据抓取！连续失败次数: " + consecutiveFailures.get());
        }
    }

    /**
     * 启动监控
     */
    private void startMonitoring() {
        monitorExecutor.scheduleAtFixedRate(() -> {
            SystemStats stats = getSystemStats();
            System.out.println("\n" + stats);

            // 动态调整并发数
            adjustConcurrency(stats);

            // 检查告警条件
            checkAlerts(stats);

        }, 5, 5, TimeUnit.SECONDS); // 每5秒监控一次
    }

    /**
     * 动态调整并发数
     */
    private void adjustConcurrency(SystemStats stats) {
        if (stats.avgResponseTime > 2000 && semaphore.availablePermits() > 1) {
            // 响应时间过长，减少并发
            semaphore.acquireUninterruptibly(1);
            System.out.println("响应时间过长，减少并发数到: " + (INITIAL_CONCURRENT_FETCHES - 1));
        } else if (stats.avgResponseTime < 500 && stats.activeConnections < MAX_CONCURRENT_FETCHES) {
            // 响应时间良好，增加并发
            semaphore.release(1);
            System.out.println("响应时间良好，增加并发数到: " + (INITIAL_CONCURRENT_FETCHES + 1));
        }
    }

    /**
     * 检查告警条件
     */
    private void checkAlerts(SystemStats stats) {
        if (stats.successRate < 80) {
            System.err.println("🚨 告警：成功率低于80%！当前成功率: " + String.format("%.1f%%", stats.successRate));
        }

        if (stats.avgResponseTime > 3000) {
            System.err.println("🚨 告警：平均响应时间超过3秒！当前: " + stats.avgResponseTime + "ms");
        }
    }

    /**
     * 获取系统统计信息
     */
    public SystemStats getSystemStats() {
        int total = successCount.get() + failureCount.get();
        long avgTime = successCount.get() > 0 ? totalFetchTime.get() / successCount.get() : 0;
        int activeConnections = INITIAL_CONCURRENT_FETCHES - semaphore.availablePermits();
        int queueSize = ((ThreadPoolExecutor) executor).getQueue().size();

        // 计算系统负载（基于队列大小和活跃连接数）
        double systemLoad = Math.min(1.0, (activeConnections + queueSize) / (double) MAX_CONCURRENT_FETCHES);

        return new SystemStats(total, successCount.get(), failureCount.get(),
                avgTime, activeConnections, queueSize, circuitBreakerOpen, systemLoad);
    }

    /**
     * 输出统计信息
     */
    private void printStatistics(long totalTime) {
        System.out.println("\n📊 高级数据抓取统计");
        System.out.println("=".repeat(60));

        SystemStats stats = getSystemStats();
        System.out.println(stats);

        System.out.printf("缓存命中: %d 次%n",
                (int) cachedResults.values().stream().filter(r -> !isCacheExpired(r)).count());
        System.out.printf("总耗时: %dms%n", totalTime);

        System.out.println("\n📋 详细结果:");
        cachedResults.values().stream()
                .sorted(Comparator.comparing(FetchResult::getFetchTime).reversed())
                .forEach(result -> System.out.println("  " + result));
    }

    /**
     * 关闭资源
     */
    public void shutdown() {
        System.out.println("\n🔄 关闭高级数据抓取系统...");
        monitorExecutor.shutdown();
        executor.shutdown();

        try {
            if (!monitorExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                monitorExecutor.shutdownNow();
            }
            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            monitorExecutor.shutdownNow();
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }

        System.out.println("高级数据抓取系统已关闭");
    }

    /**
     * 主方法
     */
    public static void main(String[] args) {
        AdvancedDataFetchDemo demo = new AdvancedDataFetchDemo();

        try {
            // 执行高级并发数据抓取
            CompletableFuture<Map<String, FetchResult>> future = demo.fetchDataConcurrently();

            // 等待完成
            Map<String, FetchResult> results = future.join();

            // 展示结果
            System.out.println("\n🎯 数据抓取完成！");
            System.out.printf("成功抓取 %d 个数据源的数据%n", results.size());

            // 模拟后续处理
            processResults(results);

        } finally {
            demo.shutdown();
        }
    }

    /**
     * 处理结果
     */
    private static void processResults(Map<String, FetchResult> results) {
        System.out.println("\n🔍 开始结果处理...");

        // 数据质量分析
        Map<String, List<FetchResult>> qualityGroups = results.values().stream()
                .collect(Collectors.groupingBy(result -> {
                    if (result.getDataQualityScore() >= 90) return "优秀";
                    if (result.getDataQualityScore() >= 70) return "良好";
                    return "一般";
                }));

        System.out.println("数据质量分布:");
        qualityGroups.forEach((quality, dataList) -> {
            System.out.printf("  %s: %d 条数据%n", quality, dataList.size());
        });

        // 生成最终报告
        generateFinalReport(results, qualityGroups);
    }

    /**
     * 生成最终报告
     */
    private static void generateFinalReport(Map<String, FetchResult> results,
                                            Map<String, List<FetchResult>> qualityGroups) {
        System.out.println("\n📈 最终报告");
        System.out.println("=".repeat(50));

        long successCount = results.values().stream().filter(FetchResult::isSuccess).count();
        double successRate = (double) successCount / results.size() * 100;

        System.out.printf("总数据源: %d%n", results.size());
        System.out.printf("成功抓取: %d (%.1f%%)%n", successCount, successRate);
        System.out.printf("平均数据质量: %.1f分%n",
                results.values().stream()
                        .filter(FetchResult::isSuccess)
                        .mapToInt(FetchResult::getDataQualityScore)
                        .average()
                        .orElse(0.0));

        System.out.println("\n✅ 高级数据抓取系统演示完成！");
    }
}